package com.atguigu.datastream.day03;

import com.atguigu.datastream.bean.Event;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.util.Collector;

import java.util.Random;

/**
 * ClassName: Flink_Source_Kafka
 * Package: com.atguigu.day03
 * Description:
 *                    用户自定义
 * @Author ChenJun
 * @Create 2023/4/8 16:40
 * @Version 1.0
 */
public class Flink_14_Source_userDefaule1 {
    public static void main(String[] args) throws Exception {
        //1.获取Flink 流的运行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);

        DataStreamSource<Event> streamSource = env.addSource(new ClickSource()).setParallelism(2);
        streamSource.process(new ProcessFunction<Event, String>() {
            @Override
            public void processElement(Event value, ProcessFunction<Event, String>.Context ctx, Collector<String> out) throws Exception {
                System.out.println("waterMark:"+ctx.timerService().currentProcessingTime());
            }
        });


        streamSource.print();


        env.execute();
    }

//    public static  class ClickSource implements SourceFunction<Event>{
    // 以下是生成并行度的方法
public static  class ClickSource implements ParallelSourceFunction<Event> {
        private Boolean isRunning = true;

        /**
         * 方法开始执行， 用于生成数据  具体的逻辑在代码中
         * @param ctx
         * @throws Exception
         */
        @Override
        public void run(SourceContext<Event> ctx) throws Exception {
            System.out.println("-------数据开始----------");
            Random random = new Random();

            String[] users = {"Songsong", "Bingbing"};
            String[] urls = {"./home", "./cart", "./fav", "./prod?id=1", "./prod?id=2"};

            while (true){
                Event element = new Event(users[random.nextInt(users.length)], urls[random.nextInt(urls.length)]
                        , System.currentTimeMillis());

                ctx.collectWithTimestamp(element,element.timestamp);

                //发送watermark
                ctx.emitWatermark(new Watermark(element.timestamp - 1));

                Thread.sleep(200);
            }
        }

        @Override
        public void cancel() {
            System.out.println("-------关闭数据-----");
         isRunning = false;
        }
    }
}
